package cn.ipanel.bigdata.dw

import cn.ipanel.bigdata.boot.config.Configuration
import cn.ipanel.bigdata.boot.date.Day
import cn.ipanel.bigdata.boot.period.Period
import cn.ipanel.bigdata.comm.Parameter.DWM_DB_NAMESPACE
import cn.ipanel.bigdata.dw
import cn.ipanel.bigdata.utils.Dictionary.NONE
import org.apache.spark.sql.functions.{col, lit}
import org.apache.spark.sql.{DataFrame, Row}

/**
 * Author: lzz
 * Date: 2022/3/25 9:01
 */
package object dwm {

  final lazy val BUILD: String = "create database if not exists " + DWM_DB_NAME
  final lazy val DWM_DB_NAME: String = Configuration.getParam(DWM_DB_NAMESPACE)

  final lazy val Tables: Seq[AbsDwmHive] = {
    Seq()
  }

  trait DwmBasicColumnDefine {
    final val F_CREATE_TIME   : String = "f_create_time"  // 创建时间
    final val F_MODIFY_TIME   : String = "f_modify_time"  // 更新时间
    final val F_PERIOD_TYPE   : String = "f_period_type"  // 周期类型（分区组合字段一）
  }

  abstract class AbsDwmHive(tbName: String,  dbName: String = DWM_DB_NAME)
          extends dwd.AbsDwdHive(tbName, dbName)
            with DwmBasicColumnDefine {

    def createTable: AbsDwmTable

    def find(day: Day, period: Period): DataFrame = {
      load.filter(col(F_PERIOD_TYPE) === lit(period.getType) and
                  col(F_DATE) === lit(period.asDate(day)))
    }

    def drop(day: Day, period: Period): Unit = {
      exec(s"""
              | alter table $getDBName.$getTBName drop if exists partition
              | ($F_PERIOD_TYPE=${period.getType}, $F_DATE=${period.asDate(day)})
              |""".stripMargin)
    }

    override def getTBColumns: Seq[String] = {
      Seq(F_DEVICE_ID, F_REGION) ++ extraColumnNames ++ Seq(F_COUNT, F_CREATE_TIME,
        F_MODIFY_TIME, F_PERIOD_TYPE, F_DATE)
    }

    override def buildTable: String = {
      s"""
         | CREATE TABLE IF NOT EXISTS $getDBName.$getTBName (
         |   $F_DEVICE_ID STRING,
         |   $F_REGION STRING,
         |   $extraColumnBuild,
         |   $F_COUNT BIGINT,
         |   $F_CREATE_TIME BIGINT,
         |   $F_MODIFY_TIME BIGINT
         | )
         | PARTITIONED BY($F_PERIOD_TYPE INT, $F_DATE INT)
         | ROW FORMAT DELIMITED FIELDS TERMINATED BY $dataRowFormatter
         | MAP KEYS TERMINATED BY $dataMapFormatter
         | STORED AS $storageFormatter
         |""".stripMargin
//      CLUSTERED BY($F_USER, $F_DEVICE) INTO $splitBucketsNums BUCKETS
    }
  }

  abstract class AbsDwmTable(f_device_id: String,
                             f_region: String,
                             f_count: Long,
                             var f_create_time: Long,
                             var f_modify_time: Long,
                             var f_period_type: Int,
                             f_date: Int)
          extends dwd.AbsDwdTable(f_date_time = NONE, f_device_id, f_region, f_count,
            f_extra = Map(), f_date) with DwmBasicColumnDefine {

    override lazy val _PARAM: Array[Any] = Array(f_device_id, f_region,
      f_count) ++ toParam ++ Array(f_create_time, f_modify_time, f_period_type, f_date)

    override def _init_(row: Row): AbsDwmTable = {
      super._init_(row)
      f_create_time = row.getAs[Long](F_CREATE_TIME)
      f_modify_time = row.getAs[Long](F_MODIFY_TIME)
      f_period_type = row.getAs[Int](F_PERIOD_TYPE)
      this
    }
  }
}